package com.ikas.autumn.iec104.module.data.mysql.handler;

import cn.hutool.cache.CacheUtil;
import cn.hutool.cache.impl.TimedCache;
import cn.hutool.core.lang.UUID;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.List;

/**
 * 消息生产处理
 * @author dbh
 */
@Slf4j
@Component
public class ProducerHandler {

    /**
     * 默认分区
     */
    private int defPartition = 0;

    /**
     * 104消息发送topic
     */
    @Value("${spring.kafka.producer.topic.104-topic}")
    private String topicOf104;


    /**
     * 总召唤发送开关
     */
    @Value("${total-summon.filter-ts-flag}")
    private Boolean filterTsFlag;

    /**
     * 保存总召唤报文开关
     */
    @Value("${total-summon.save-ts-code-flag}")
    private Boolean saveTsCodeFlag;
    /**
     * 保存总召唤数据开关
     */
    @Value("${total-summon.save-ts-data-flag}")
    private Boolean saveTsDataFlag;

    /**
     * 接收数据保存开关
     */
    @Value("${total-summon.save-receive-data-flag}")
    private Boolean saveReceiveDataFlag;

    /**
     * 接收遥信数据保存开关
     */
    @Value("${total-summon.save-yx-receive-data-flag}")
    private Boolean saveYxReceiveDataFlag;

    /**
     * 接收遥测数据保存开关
     */
    @Value("${total-summon.save-yc-receive-data-flag}")
    private Boolean saveYcReceiveDataFlag;

    /**
     * 遥信遥测本地缓存时间 （位：毫秒）
     */
    @Value("${local-cache-timeout:3600000}")
    private Long localCacheTimeout;

    /**
     * 遥信本地缓存
     */
    private TimedCache<String, List<Long>> yxLocalCache;
    /**
     * 遥测本地缓存
     */
    private TimedCache<String, List<Long>> ycLocalCache;


    @Resource
    private KafkaTemplate<String,Object> kafkaTemplate;

    /**
     * 获取过滤总召唤发送开关信息
     * @return Boolean
     */
    public Boolean getFilterTsFlag() {
        return filterTsFlag;
    }

    /**
     * 获取保存总召唤报文开关信息
     * @return Boolean
     */
    public Boolean getSaveTsCodeFlag() {
        return saveTsCodeFlag;
    }

    /**
     * 获取保存总召唤数据开关信息
     * @return Boolean
     */
    public Boolean getSaveTsDataFlag() {
        return saveTsDataFlag;
    }

    /**
     * 获取保存接收数据开关信息
     * @return Boolean
     */
    public Boolean getSaveReceiveDataFlag() {
        return saveReceiveDataFlag;
    }

    /**
     * 获取保存遥信接收数据开关信息
     * @return Boolean
     */
    public Boolean getSaveYxReceiveDataFlag() {
        return saveYxReceiveDataFlag;
    }

    /**
     * 获取保存遥测接收数据开关信息
     * @return Boolean
     */
    public Boolean getSaveYcReceiveDataFlag() {
        return saveYcReceiveDataFlag;
    }
    /**
     * 获取遥信遥测本地缓存时间 （位：毫秒）
     * @return Boolean
     */
    public TimedCache<String, List<Long>> getYxLocalCache() {
        if (yxLocalCache == null) {
            yxLocalCache = CacheUtil.newTimedCache(localCacheTimeout);
        }
        return yxLocalCache;
    }

    /**
     * 获取遥信遥测本地缓存时间 （位：毫秒）
     * @return Boolean
     */
    public TimedCache<String, List<Long>> getYcLocalCache() {
        if (ycLocalCache == null) {
            ycLocalCache = CacheUtil.newTimedCache(localCacheTimeout);
        }
        return ycLocalCache;
    }

    /**
     * 加入kafka消息队列
     * @param value 对象值
     */
    public void sendYxYcKafka(Object value) {
        kafkaTemplateSend(UUID.randomUUID().toString(), value, defPartition, topicOf104);
    }

    /**
     * 加入kafka消息队列
     * @param key key of the datastore
     * @param value 对象值
     */
    public void sendYxYcKafka(String key, Object value) {
        kafkaTemplateSend(key, value, defPartition, topicOf104);
    }

    private void kafkaTemplateSend(String key, Object value, Integer partition, String topic) {
        ProducerRecord<String, Object> producerRecord = new ProducerRecord<>(
                topic, partition, null, key, value
        );
        kafkaTemplate.send(producerRecord).addCallback(
                new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                // 发送失败的处理
                log.info("mt-104-kafka send topic[{}] 生产者发送消息失败[{}]", topic, throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                // 成功的处理
                log.info("mt-104-kafka send topic[{}],partition:{} 生产者发送消息成功 key:{}", topic, partition,stringObjectSendResult.getProducerRecord().key());
            }
        });
    }

}
